package util;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;

import java.io.IOException;
import java.io.StringReader;
import java.nio.ByteBuffer;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.KeyRange;
import org.apache.cassandra.thrift.KeySlice;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.lucene.analysis.LowerCaseFilter;
import org.apache.lucene.analysis.PorterStemFilter;
import org.apache.lucene.analysis.standard.StandardTokenizer;
import org.apache.lucene.analysis.tokenattributes.CharTermAttribute;
import org.apache.lucene.util.Version;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.TException;


/**
 * @author Stanley
 *
 */
public class KeyWordSearchService {
	String keyspaceName;
	String columnFamily;
	int numRowKeys;

	public KeyWordSearchService(String keyspaceName, String columnFamily) {
		this.keyspaceName = keyspaceName;
		this.columnFamily = columnFamily;
	}

	public int getNumRowKeys() throws InvalidRequestException, TException, UnavailableException, TimedOutException {

		TTransport tr = new TFramedTransport(new TSocket("localhost", 9160));
		TProtocol proto = new TBinaryProtocol(tr);

		Cassandra.Client client = new Cassandra.Client(proto);
		tr.open();
		String keyspace = keyspaceName;
		client.set_keyspace(keyspace);
		// record id
		ColumnParent columnParent = new ColumnParent(columnFamily);

		SlicePredicate predicate = new SlicePredicate();
		predicate.setSlice_range(new SliceRange(ByteBuffer.wrap(new byte[0]),
				ByteBuffer.wrap(new byte[0]), false, 200));

		KeyRange keyRange = new KeyRange(200);
		keyRange.setStart_key(new byte[0]);
		keyRange.setEnd_key(new byte[0]);
		List<KeySlice> keySlices = client.get_range_slices(columnParent,
				predicate, keyRange, ConsistencyLevel.ONE);

		tr.close();
		
		numRowKeys = keySlices.size();
	    
		return numRowKeys;
	}
	
	@SuppressWarnings("deprecation")
	public String stemUserInput(String toStem) throws IOException {
		
		StandardTokenizer stdTokenizer = new StandardTokenizer(
				Version.LUCENE_35, new StringReader(toStem));
		
		LowerCaseFilter lcFilter = new LowerCaseFilter(stdTokenizer);
		
		PorterStemFilter psFilter = new PorterStemFilter(lcFilter);
		
		psFilter.incrementToken();
		CharTermAttribute charTermAttribute = psFilter
				.getAttribute(CharTermAttribute.class);
		String term = charTermAttribute.toString();
		
		return term;
		
	}
	
	@SuppressWarnings({ "rawtypes", "unchecked" })
	public LinkedHashMap searchUserKeyWord(CassandraService cs, int numRowKeys, String keyword, int min, int max) throws InvalidRequestException, TException, UnavailableException, TimedOutException {
		String rowKey = "html_";
		
		// keep min<value<max & it's htm_url
		HashMap<String, Integer> unSorted = new HashMap<String, Integer>();
		
		unSorted = cs.queryHtmlData(columnFamily, numRowKeys, rowKey, keyword);
		   
	    // sort by descending hits
		List mk = new ArrayList(unSorted.keySet());
		List mv = new ArrayList(unSorted.values());
		Collections.sort(mv, Collections.reverseOrder());
		Collections.sort(mk, Collections.reverseOrder());

		LinkedHashMap sortedMap = new LinkedHashMap();

		Iterator it = mv.iterator();
		while (it.hasNext()) {
			Object value = it.next();
			Iterator kIt = mk.iterator();

			while (kIt.hasNext()) {
				Object key = kIt.next();
				String str1 = unSorted.get(key).toString();
				String str2 = value.toString();

				if (str1.equals(str2)) {
					unSorted.remove(key);
					mk.remove(key);
					sortedMap.put((String) key, (Integer) value);
					break;
				}
			}
		} // end of outer most while
		
		return sortedMap;
	}
}
